package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.DimJoinFunction;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.HbaseUtil;
import com.atguigu.gmall.realtime.utils.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2023/11/21
 * 抽取维度关联的模板类
 * 模板方法设计模式：在父类(模板)中定义实现某一个功能的核心算法骨架(步骤)，某些步骤没有办法在父类中完成，可以定义成
 *              抽象的，将实现延迟到子类中去做。
 *              好处：在不改变父类核心算法估计的前提下，每个子类都可以有自己不同的实现
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T> {

    private Connection hbaseConn;
    private Jedis redisConn;
    private String tableName;

    public DimMapFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HbaseUtil.getHbaseConnection();
        redisConn = RedisUtil.getRedisConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(hbaseConn);
        RedisUtil.closeRedisConnection(redisConn);
    }

    @Override
    public T map(T obj) throws Exception {
        //根据流中的对象获取要关联的维度主键
        String key = getKey(obj);
        //根据维度主键获取维度对象
        JSONObject dimInfoJsonObj = DimUtil.getDimInfo(redisConn, hbaseConn, GmallConfig.HBASE_NAMESPACE, tableName, key);
        //将维度对象相关的维度属性补充到流中的对象上
        if(dimInfoJsonObj != null){
            join(dimInfoJsonObj,obj);
        }
        return obj;
    }
}
